package com.atguigu.chapter07.D01_Window;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Author: Pepsi
 * Date: 2023/8/4
 * Desc:
 */
public class Flink05_Window_NoKey {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        env
                .socketTextStream("hadoop101", 9999)
                .map(line -> {
                    String[] data = line.split(",");

                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum("vc")  //窗口处理的并行度必须是：1
                .print();
/*
* 全部：所有的元素都会进入到同一个窗口
*       比如：0-5 所有的元素进入同一个窗口
*            5-10
* */
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}